package com.lagou.kafka.demo.serializer;

import com.lagou.kafka.demo.entity.User;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description:  自定义序列化器
 * @date 2022/6/4 11:17
 */
public class UserSerializer implements Serializer<User> {

    public void configure(Map<String, ?> configs, boolean isKey) {
        // todo 用于接收序列化器的配置参数，并对当前序列化器进行配置和初始化
    }

    /**
     * @description: 使用nio的方式来进行序列化
     * @author oyz
     * @param topic
     * @param user
     * @date 2022/8/17 21:09
     * @version 1.0
     */
    public byte[] serialize(String topic, User user) {
        try{
            if(null == user){
                return new byte[0];
            }
            Integer userId = user.getUserId();
            String userName = user.getUserName();
            int length = 0;
            byte[] bytes = null;
            if(null != userName){
                bytes = userName.getBytes("UTF-8");
                length = bytes.length;
            }
            // 第一个4用于存放userId的int类型，占4个字节
            // 第二个4用于存放length的int类型，占4个字节
            // 第三个长度用于存放username序列化后的字节数组
            ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + length);
            // 设置userId
            buffer.putInt(userId);
            // 设置userName字节数组长度
            buffer.putInt(length);
            // 设置userName字节数组
            buffer.put(bytes);
            return buffer.array();

        }catch (Exception e){
            throw new SerializationException("序列化异常");
        }

    }

    public byte[] serialize(String topic, Headers headers, User data) {
        return new byte[0];
    }

    public void close() {
        // todo 用于关闭资源操作，需要幂等，多次调用效果是一样的
    }
}
